package service

import (
	"gitee.com/crack007/goose"
	"github.com/Shopify/sarama"
	"github.com/crack007/go-im/src/common/config"
)

type KafkaProducer struct {
	client sarama.SyncProducer
}

func NewKafkaProducer() *KafkaProducer {
	return &KafkaProducer{}
}

func (k *KafkaProducer) Init() error {
	var err error
	k.client, err = sarama.NewSyncProducer(config.KafkaConfig.Host, nil)
	if err == nil {
		goose.GetLogger().Debug("Kafka connected")
	}
	return err
}
func (k *KafkaProducer) Close() {
	k.client.Close()
}

func (k *KafkaProducer) Publish(topic string, data string) error {
	goose.GetLogger().Debug("kafka publish-> topic:%s data:%s", topic, data)
	msg := &sarama.ProducerMessage{}
	msg.Topic = topic
	msg.Value = sarama.StringEncoder(data)
	_, _, err := k.client.SendMessage(msg)
	return err
}
